package main

import (
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	inf "github.com/fzdwx/infinite"
	"github.com/fzdwx/infinite/components/input/text"
	"github.com/fzdwx/infinite/theme"
)

// 节点信息
type BrokerInfo struct {
	// 节点Ip:端口
	Servers []string
	// 消息主题
	Topic string
	// 分区
	Partition int32
	// 最大重试次数
	maxRetry int
}

var broker = &BrokerInfo{}

func createText(tip, defaultVal string) *text.Text {
	return inf.NewText(
		text.WithRequired(),
		text.WithPrompt(tip),
		text.WithPromptStyle(theme.DefaultTheme.PromptStyle),
		text.WithDefaultValue(defaultVal),
	)
}

func main() {
	title := `
	_          __ _                               _                     
	| |        / _| |                             | |                    
	| | ____ _| |_| | ____ _   _ __  _ __ ___   __| |_   _  ___ ___ _ __ 
	| |/ / _  |  _| |/ / _  | | '_ \| '__/ _ \ / _  | | | |/ __/ _ \ '__|
	|   < (_| | | |   < (_| | | |_) | | | (_) | (_| | |_| | (_|  __/ |   
	|_|\_\__,_|_| |_|\_\__,_| | .__/|_|  \___/ \__,_|\__,_|\___\___|_|   
							  | |                                        
							  |_|                                        
	
	                             exit/quit 
	`
	fmt.Println(title)
	tip("servers", "please input ip and port (eg: host1:9092,host2:9092)", 0, handleServers)
	tip("topic", "please input topic", 0, handleTopic)
	tip("partition", "please input partition(eg: 0)", 0, handlePartition)
	tip("max retry", "please input pmax retry(eg: 5)", 0, handleMaxRetry)

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = broker.maxRetry
	config.Producer.Return.Successes = true
	producer, err := sarama.NewSyncProducer(broker.Servers, config)
	if err != nil {
		log.Fatalf("connect error: %#v", err.Error())
	}

	defer func() {
		closeProducer(producer)
	}()

	count := 0
	for {
		send(func(msg string) error {
			switch msg {
			case "exit", "quit":
				closeProducer(producer)
				os.Exit(0)
			}
			_, _, err := producer.SendMessage(&sarama.ProducerMessage{
				Topic:     broker.Topic,
				Key:       nil,
				Value:     sarama.StringEncoder(msg),
				Partition: broker.Partition,
			})
			return err
		}, count)
	}
}

func handleServers(servers string) error {
	broker.Servers = strings.Split(servers, ",")
	return nil
}

func handlePartition(partition string) error {
	part, err := strconv.Atoi(partition)
	if err != nil {
		return err
	}

	broker.Partition = int32(part)
	return nil
}

func handleTopic(topic string) error {
	broker.Topic = topic
	return nil
}

func handleMaxRetry(maxRetry string) error {
	retry, err := strconv.Atoi(maxRetry)
	if err != nil {
		return err
	}

	broker.maxRetry = retry
	return nil
}

func send(handle func(msg string) error, count int) {
	tip := "please input message:"
	if count > 0 {
		tip = "please input next message:"
	}
	value, _ := createText(tip, "").Display()
	handle(value)
}

func closeProducer(producer sarama.SyncProducer) {
	log.Println("close producer")
	if err := producer.Close(); err != nil {
		log.Panic(err)
	}
}

func tip(message, defaultVal string, count int, handleFun func(value string) error) {
	if count > 2 {
		fmt.Printf("too many errors(%d times), system will exit after 3 second.\n", count)
		time.Sleep(time.Second * 2)
		os.Exit(0)
	}
	if count > 0 {
		fmt.Printf("input error(%d times), retry please……\n", count)
	}
	value, err := createText(message, defaultVal).Display()
	if err != nil {
		tip(message, defaultVal, count+1, handleFun)
	}
	switch value {
	case "exit", "quit":
		os.Exit(0)
	}
	if err = handleFun(value); err != nil {
		tip(message, defaultVal, count+1, handleFun)
	}
}
